package cn.wanda.utils.modelGroup

import java.lang._
import cn.wanda.utils.config.groupJobTrait
import org.apache.spark.sql.{Row, SparkSession}

class createAvgShiDong extends groupJobTrait {

    /**
      * shidong_avg_3mon
      */
    def create(sparkSessionContext: SparkSession): Unit = {
        // 求出所有用户还款列表
        val tran_amt_paid_list = sparkSessionContext.sql(shiDongPaidSql)
          .rdd
          .groupBy(t => t.get(0))
          .mapValues(_.map(x => (x.get(1), x.get(3), x.get(2), "paid")))

        // 求出所有用户借款列表
        val posting_list = sparkSessionContext.sql(shiDongLoanSql)
          .rdd
          .groupBy(t => t.get(0))
          .mapValues(_.map(x => (x.get(1), x.get(3), x.get(2), "posting")))

        val shiDongMap = tran_amt_paid_list.union(posting_list)
          .reduceByKey(_ ++ _)
          .map(f => {
              // member_code
              val memberCode = f._1.toString

              // payment信息
              val paymentInfo = f._2

              // 借款总金额
              var postingSum90 = 0.0
              paymentInfo.filter(_._4 == "posting").map(_._2).toList.foreach(t => {
                  postingSum90 += Double.valueOf(t.toString)
              })

              // 还款总金额
              var paidSum90 = 0.0
              paymentInfo.filter(_._4 == "paid").map(_._2).toList.foreach(t => {
                  paidSum90 += Double.valueOf(t.toString)
              })

              // 额度
              val cashAtm = paymentInfo.map(_._3).toList.head.toString

              // 当天实动率
              val nowDateShiDong = (postingSum90 - paidSum90) / cashAtm.toDouble

              // 总列表，已处理借款和还款,只会存在一天，并且借款减去还款
              val totalList = paymentInfo.groupBy(_._1).mapValues(x => {
                  if (x.size == 2) {
                      if (x.head._4 == "paid") {
                          val todayMem:Float = x.takeRight(1).head._2.toString.toFloat - x.head._2.toString.toFloat
                          (todayMem, x.head._3)
                      } else {
                          val todayMem:Float = x.head._2.toString.toFloat - x.takeRight(1).head._2.toString.toFloat
                          (todayMem, x.head._3)
                      }
                  } else {
                      (x.head._2, x.head._3)
                  }
              })

              // 总的日期(包含借还款)
              val allDate = totalList.keys.toList.sortBy(x => dateFormatByDay.parse(x.toString).getTime)

              // 最早记录时间
              val minDate = allDate.head.toString

              // 求90天内是否有记录，如果不满90天，那么按照起始时间开始计算
              val endDate90 = if (diffDay(NowDate(), minDate) < 90) diffDay(NowDate(), minDate) else 90

              // 过滤出限定期间内的有过发生交易的时间节点
              var paymentDateEventList = {
                  allDate.filter(x => {
                      diffDay(NowDate(), x.toString) <= endDate90 && diffDay(NowDate(), x.toString) >= 0
                  })
              }

              var shiDong = 0.0

              if (paymentDateEventList.isEmpty) {
                  // 代表规定时间内没有发生任何借款或还款，所以只需要计算现在总实动率，就是90天内的平均实动率
                  shiDong = nowDateShiDong
              } else {
                  // 目前总金额(借款金额-还款金额)
                  var amountTmp = postingSum90 - paidSum90

                  // 某一区间内的实动率，因为从后向前算，所以初始值为当前实动率
                  var shiDongTmp1 = nowDateShiDong

                  // 累加的实动率
                  var shiDongTmp2 = 0.0

                  // 后一个时间，初始值为当前
                  var afterDateTmp = NowDate()

                  // 前一个时间
                  var beforeDayTmp = ""

                  for (i <- paymentDateEventList.indices) {
                      // 取出相近的那一天，并从列表内删除
                      beforeDayTmp = paymentDateEventList.takeRight(1).head.toString

                      // 从列表内删除最后一个日期
                      paymentDateEventList = paymentDateEventList.dropRight(1)

                      // 相差天数
                      val diffDateNum = diffDay(afterDateTmp, beforeDayTmp)

                      // 计算区间实动率(没有变化的时候，就是天数乘以当前实动率)
                      shiDongTmp2 += shiDongTmp1 * diffDateNum

                      // 更新前一天时间
                      afterDateTmp = beforeDayTmp

                      // 更新目前实动率分子
                      val nowDayMen: Float = totalList(beforeDayTmp).asInstanceOf[(Float, Float)]._1.asInstanceOf[Float]

                      amountTmp -= nowDayMen

                      // 更新实动率
                      shiDongTmp1 = amountTmp / cashAtm.toDouble
                  }

                  if (diffDay(NowDate(), beforeDayTmp) < endDate90) {
                      // 计算剩余差额天数
                      val diff = endDate90 - diffDay(NowDate(), beforeDayTmp)
                      shiDongTmp2 += shiDongTmp1 * diff
                  }
                  shiDong = shiDongTmp2 / endDate90
              }
              (memberCode, shiDong)
          })

        val shiDongRdd = shiDongMap.map(
            x => Row(x._1, x._2)
        )

        sparkSessionContext.sqlContext
          .createDataFrame(shiDongRdd, shidongSchema)
          .createTempView("shidong_avg_3mon")

    }
}
